import rx.Observable;
import rx.Subscription;
import rx.observables.ConnectableObservable;

import java.io.IOException;
import java.util.concurrent.TimeUnit;

/**
 * Created by codeczx on 2017/2/24.
 */
public class ColdAndHotObservableTest {

    void testHotObservable() throws InterruptedException, IOException {
        ConnectableObservable<Long> connectable = Observable.interval(200,TimeUnit.MILLISECONDS).publish();
        connectable.connect();
        Thread.sleep(1000);
        connectable.subscribe(i->{
            System.out.println("first:"+i);
        });
        Thread.sleep(1000);
        connectable.subscribe(i->{
            System.out.println("second:"+i);
        });
        System.in.read();
    }

    void testPublish() throws InterruptedException, IOException {
        ConnectableObservable<Long> cold = Observable.interval(200, TimeUnit.MILLISECONDS).publish();
        cold.connect();
        Thread.sleep(1000);
        cold.subscribe(i -> System.out.println("First: " + i));
        Thread.sleep(500);
        cold.subscribe(i -> System.out.println("Second: " + i));
        System.in.read();
    }

    void testDisConnecting() throws InterruptedException, IOException {
        ConnectableObservable<Long> connectable = Observable.interval(200, TimeUnit.MILLISECONDS).publish();
        Subscription s = connectable.connect();

        connectable.subscribe(i -> System.out.println(i));

        Thread.sleep(1000);
        System.out.println("Closing connection");
        s.unsubscribe();

        Thread.sleep(1000);
        System.out.println("Reconnecting");
        s = connectable.connect();
        connectable.subscribe(i -> System.out.println(i));
        System.in.read();
    }

    void testReplay() throws InterruptedException, IOException {
        ConnectableObservable<Long> cold = Observable.interval(200,TimeUnit.MILLISECONDS).take(10).replay();
        Subscription s = cold.connect();
        cold.subscribe(i->System.out.println("first:"+i));
        Thread.sleep(1000);
        cold.subscribe(i->System.out.println("second:"+i));
        System.in.read();
    }

    void testCache(){
        Observable<Long> values = Observable.interval(200,TimeUnit.MILLISECONDS)
                .take(5)
                .cache();
        values.subscribe(i->System.out.println("first:"+i));
        values.subscribe(i->System.out.print("second:"+i));
    }

    public static void main(String[] args) throws InterruptedException, IOException {
        ColdAndHotObservableTest cahot = new ColdAndHotObservableTest();
        cahot.testReplay();
//        cahot.testPublish();
//        cahot.testDisConnecting();
//        cahot.testHotObservable();
//        try {
//            System.in.read();
//        } catch (IOException e) {
//            e.printStackTrace();
//        }
    }
}
